#include <cserver/session/node.h>
#include <boost/asio.hpp>
#include <cserver/context.h>
#include <boost/log/trivial.hpp>
#include <xirang2/utility/unused.h>
#include <boost/lexical_cast.hpp>
#include <chrono>
#include <ctime>
#include <cryptopp/aes.h>
#include <cryptopp/filters.h>
#include <cryptopp/modes.h>



namespace csrv{
	using namespace boost::asio::ip;

	template<typename Fun> bool catch_call(Fun f){
		try{
			return f();
		}catch(std::exception& e){
			BOOST_LOG_TRIVIAL(warning) <<"Process data failed: " << e.what();
		}
		catch(...){
			BOOST_LOG_TRIVIAL(warning) <<"Process data failed: unknown exception";
		}
		return false;
	}

#if 1
	template<typename Ar, typename =
		typename std::enable_if<s11n::is_deserializer<Ar>::value>::type>
		Ar& load (Ar& ar, fire_time& v)
		{
			char* first = reinterpret_cast<char*>(&v);
			char* last = first + sizeof(v);
			block_read(ar.get(), make_range(first, last));
			return ar;
		}
	template<typename Ar, typename =
		typename std::enable_if< s11n::is_serializer<Ar>::value>::type>
		Ar& save(Ar& ar, const fire_time& v)
		{
			const char* first = reinterpret_cast<const char*>(&v);
			const char* last = first + sizeof(v);
			block_write(ar.get(), make_range(first, last));
			return ar;
		}
	template<typename Ar, typename T, size_t N, typename =
		typename std::enable_if<s11n::is_deserializer<Ar>::value>::type>
		Ar& load (Ar& ar, std::array<T, N>& v)
		{
			char* first = reinterpret_cast<char*>(v.begin());
			char* last = first + sizeof(v);
			block_read(ar.get(), make_range(first, last));
			return ar;
		}
	template<typename Ar, typename T, size_t N, typename =
		typename std::enable_if< s11n::is_serializer<Ar>::value>::type>
		Ar& save(Ar& ar, const std::array<T, N>& v)
		{
			const char* first = reinterpret_cast<const char*>(v.begin());
			const char* last = first + sizeof(v);
			block_write(ar.get(), make_range(first, last));
			return ar;
		}
#endif



	const char hex_table[]="0123456789abcdef";
	std::string hex_encode(const char* buf, size_t n){
		std::string res;
		res.resize(n * 2);
		auto dest = res.begin();
		for(const char* itr(buf), *end(buf + n); itr != end; ++itr) {
			auto idx = uint8_t(*itr);
			*dest++ = hex_table[(idx >> 4) & 0x0f];
			*dest++ = hex_table[idx & 0x0f];
		}
		return res;
	}
	std::string to_hex_string(const std::vector<char>& buf, size_t n){
		std::stringstream os;
		os << "[ ";
		for(size_t i = 0; i < n; i++)
			os << std::hex << std::setfill('0') << std::setw(2) << uint32_t(uint8_t(buf[i])) << " ";
		os << "]";
		return os.str();
	}



	NodeProtocol::NodeProtocol(Context& ctx) 
		: context(ctx)
		, socket_forward(ctx.ios_aserver, udp::endpoint(udp::v4(), 0))
	{}

	NodeTcp::NodeTcp(Context& ctx, std::shared_ptr<boost::asio::ip::tcp::socket> s) : NodeProtocol(ctx), socket(s){}
	void NodeTcp::start(boost::asio::yield_context yield){
		boost::system::error_code ec;
		std::vector<char> buffer;
		auto end_point = socket->remote_endpoint(ec);
		if (ec) return;

		std::string ip_port_str = boost::lexical_cast<std::string>(end_point.address()) + std::string(":") +  boost::lexical_cast<std::string>(end_point.port());
		while(socket->is_open()){
			HeadData data;

			boost::asio::async_read(*socket, boost::asio::buffer(data.buffer), yield[ec]);
			if (ec)	return;

			auto head_type = checkHead(data);
			switch(head_type){
				case HeadType::nbiot: 
					{
						if (data.head.length == 0) continue;

						buffer.resize(data.head.length);

						if (!buffer.empty()){
							boost::asio::async_read(*socket, boost::asio::buffer(buffer), yield[ec]);
							if (ec) return;
						}
						if (ec) {
							BOOST_LOG_TRIVIAL(warning) <<"NodeTCP receive: " << ec.message();
							return;
						}

						buffer.insert(buffer.begin(), std::begin(data.buffer), std::end(data.buffer));
						BOOST_LOG_TRIVIAL(debug) <<"NodeTCP data: " << to_hex_string(buffer, buffer.size());

						const char* first = &*buffer.begin();
						auto size = buffer.size();

						process(ip_port_str, data.head, first + sizeof(HeadProtocol), size - sizeof(HeadProtocol)
								, [this](const std::vector<char>& data){
								auto buf = std::make_shared<std::vector<char>>(data);
								BOOST_LOG_TRIVIAL(debug) <<"NodeTCP response: " << to_hex_string(data, data.size());
								if (socket->is_open())
									boost::asio::async_write(*socket, boost::asio::buffer(*buf), [buf](const boost::system::error_code&, std::size_t ){});
								});
					}
					break;
				case HeadType::fire:
					{
						buffer.resize(17);  // till offset 27, command byte
						boost::asio::async_read(*socket, boost::asio::buffer(buffer), yield[ec]);
						if (ec) return;

						buffer.insert(buffer.begin(), std::begin(data.buffer), std::end(data.buffer));
						auto data_size = *reinterpret_cast<uint16_t*>(&*buffer.begin() + 24); // 24: offset of data size

						buffer.resize(buffer.size() + data_size + 3);
						boost::asio::async_read(*socket, boost::asio::buffer(&*buffer.begin() + 27, data_size + 3), yield[ec]);
						if (ec) return;

						BOOST_LOG_TRIVIAL(debug) <<"NodeTCP data: " << to_hex_string(buffer, buffer.size());

						catch_call([&]{ process_fire(ip_port_str, data.head_fire, &*buffer.begin(), buffer.size(),
										[this](const std::vector<char>& data){
										auto buf = std::make_shared<std::vector<char>>(data);
										BOOST_LOG_TRIVIAL(debug) <<"NodeTCP response: " << to_hex_string(data, data.size());
										if (socket->is_open()) try{
											boost::asio::async_write(*socket, boost::asio::buffer(*buf), [buf](const boost::system::error_code&, std::size_t ){});
											}catch(std::exception& e){
												BOOST_LOG_TRIVIAL(error) <<"Failed to send NodeTCP response: " << e.what();
											}catch(...){
												BOOST_LOG_TRIVIAL(error) <<"Failed to send NodeTCP response ";
											}
										});
									return true;
								});
					}
					break;
				default:
					return;
			};
		}
	}
	void NodeTcp::run_(boost::asio::yield_context yield){
	}

	NodeUdp::NodeUdp(Context& ctx) : NodeProtocol(ctx){
	}
	void NodeUdp::start(){
		BOOST_LOG_TRIVIAL(info) << "Node UDP started";
		boost::asio::spawn(context.ios_client, [&](boost::asio::yield_context yield) {
				for(;;){
					try{ run_(yield); return; }catch(...){}
				}
				});
		boost::asio::spawn(context.ios_client, [&](boost::asio::yield_context yield) { 
				for(;;){
					try{ run_xhs_(yield); return;} catch(...){}
				}
				});
	}
	void NodeUdp::run_xhs_(boost::asio::yield_context yield){
		std::vector<char> buffer(2048);
		boost::system::error_code ec;
		udp::socket socket(context.ios_client, udp::endpoint(address::from_string(context.config.node_addr), context.config.xhs_port)); 
		for(;socket.is_open();){
			udp::endpoint sender_endpoint_;
			auto size = socket.async_receive_from(boost::asio::buffer(buffer, buffer.size()), sender_endpoint_,yield[ec]);
			if (ec) 
				BOOST_LOG_TRIVIAL(warning) <<"NodeUdp XHS receive: " << ec.message();
			else if (size > 0){
				BOOST_LOG_TRIVIAL(debug) <<"NodeUdp XHS data: " << to_hex_string(buffer, size);

				uint16_t payload_size = uint16_t(2 // property id, 0x32
						+ 2 		// message size
						+ size
						);

				std::vector<char> fake_head;
				buffer_out wr(fake_head);
				auto ar = s11n::as_sink(wr);
				ar & K_signature & K_version & K_reserved 
					& uint16_t(0x11) // XHS
					& payload_size
					& uint16_t(0x32)
					& uint16_t(size)
					;

				std::string ip_port_str = boost::lexical_cast<std::string>(sender_endpoint_.address()) 
					+ std::string(":") +  boost::lexical_cast<std::string>(sender_endpoint_.port());
				buffer.insert(buffer.begin(), fake_head.begin(), fake_head.end());
				size += fake_head.size();
				const auto& h = *reinterpret_cast<HeadData*>(&*buffer.begin());
				process(ip_port_str, h.head, &*buffer.begin() + sizeof(HeadProtocol), size - sizeof(HeadProtocol),
						[&socket, sender_endpoint_](const std::vector<char>& data){
						auto buf = std::make_shared<std::vector<char>>(data);
						if (socket.is_open())
							socket.async_send_to(boost::asio::buffer(*buf), sender_endpoint_, [buf](const boost::system::error_code&, std::size_t ){});
						});
			}
		}

	}
	void NodeUdp::run_(boost::asio::yield_context yield){
		std::vector<char> buffer(2048);
		boost::system::error_code ec;
		udp::socket socket(context.ios_client, udp::endpoint(address::from_string(context.config.node_addr), context.config.node_port)); 
		for(;socket.is_open();){
			udp::endpoint sender_endpoint_;
			auto size = socket.async_receive_from(boost::asio::buffer(buffer, buffer.size()), sender_endpoint_,yield[ec]);
			if (ec) 
				BOOST_LOG_TRIVIAL(warning) <<"NodeUdp receive: " << ec.message();
			else if (size > 0){
				BOOST_LOG_TRIVIAL(debug) <<"NodeUdp data: " << to_hex_string(buffer, size);
				if (size < sizeof(HeadData)) continue;

				std::string ip_port_str = boost::lexical_cast<std::string>(sender_endpoint_.address()) + std::string(":") +  boost::lexical_cast<std::string>(sender_endpoint_.port());
				const auto& h = *reinterpret_cast<HeadData*>(&*buffer.begin());
				auto head_type = checkHead(h);
				switch (head_type){
					case HeadType::nbiot:
						{
							if (sizeof(HeadProtocol) + h.head.length <= size){
								process(ip_port_str, h.head, &*buffer.begin() + sizeof(HeadProtocol), size - sizeof(HeadProtocol),
										[&socket, sender_endpoint_](const std::vector<char>& data){
										auto buf = std::make_shared<std::vector<char>>(data);
										if (socket.is_open())
											socket.async_send_to(boost::asio::buffer(*buf), sender_endpoint_, [buf](const boost::system::error_code&, std::size_t ){});
										});
							}
							else
								BOOST_LOG_TRIVIAL(debug) <<"NodeUdp data nbiot size mismatch: " << size << " != " << h.head.length << " + " <<  sizeof(HeadProtocol);
						}
						break;
					case HeadType::fire:
						{
							BOOST_LOG_TRIVIAL(debug) <<"NodeUdp data fire: " << to_hex_string(buffer, size);
							catch_call([&]{ process_fire(ip_port_str, h.head_fire, &*buffer.begin(), size,
										[&socket, sender_endpoint_](const std::vector<char>& data){
										auto buf = std::make_shared<std::vector<char>>(data);
										BOOST_LOG_TRIVIAL(debug) <<"NodeUdp response fire: " << to_hex_string(data, data.size());
										if (socket.is_open())
											socket.async_send_to(boost::asio::buffer(*buf), sender_endpoint_, [buf](const boost::system::error_code&, std::size_t ){});
										});
									return true;
									});
						}
						break;
					default:
						continue;
				}
			}
		}
	}


	void NodeProtocol::process_fire(const std::string& ip_addr_port, const HeadProtocolFire& head, const char* buf, size_t size, ResponseSenderType sender){
		fire_package pkg;
		pkg.seq = head.seq;
		auto rng = range<const char*>(buf + 6, buf + size);
		buffer_in rd(rng); // skip to time label
		auto ar = s11n::as_source(rd);
		ar & pkg.time;

		block_read(rd, to_range(pkg.raw_src_addr));
		block_read(rd, to_range(pkg.raw_dest_addr));
		pkg.src_addr = hex_encode(pkg.raw_src_addr.begin(), 6);
		pkg.dest_addr = hex_encode(pkg.raw_dest_addr.begin(), 6);

		auto data_size = load<uint16_t>(ar);
		ar & pkg.command;
		pkg.data.resize(data_size);
		block_read(rd, to_range(pkg.data));

		if (pkg.command != 2 && pkg.command != 5)
			return;	// 2: send data, 5: response

		bool is_ok = pkg.data.empty() 
			; // || context.database->save_fire_protection(pkg, sender);
		// response
		std::vector<char> res2;
		buffer_out wr2(res2);
		auto ar2 = s11n::as_sink(wr2);
		ar2 & K_sig_fire & pkg.seq & head.major_version & head.user_version
			& pkg.time
			& pkg.raw_dest_addr
			& pkg.raw_src_addr
			& uint16_t(0)		// data size
			& uint8_t(is_ok? 3 : 6)		// confirm command
			& uint8_t(0)
			& K_sig_fire_end;

		unsigned int sum = 0;
		for(size_t i = 2; i < 27; ++i)
			sum += res2[i];
		res2[27] = uint8_t(sum & 0xff);
		sender(res2);

		if (!this->context.config.default_forward_host.empty()){
			std::vector<char> package(buf, buf + size);
			auto self = shared_from_this();
			this->context.ios_aserver.post([self, package]{
					udp::resolver resolver(self->context.ios_aserver);
					udp::resolver::query query(udp::v4(), self->context.config.default_forward_host, self->context.config.default_forward_port);
					boost::system::error_code ec;
					auto itr = resolver.resolve(query, ec);
					if (!ec) {
					self->socket_forward.send_to(boost::asio::buffer(package, package.size()), *itr);
					BOOST_LOG_TRIVIAL(debug) <<"Forward data to " << self->context.config.default_forward_host;
					}
					});

		}
	}
	void NodeProtocol::process(const std::string& ip_addr_port, const HeadProtocol& head, const char* first, size_t size, ResponseSenderType sender){
		//assert(size <= buf.size());
		BOOST_LOG_TRIVIAL(debug) <<"process data: " << std::hex << head.cmd << ", " << head.length << "(acture " << size << ")";
		bool ok = false;
		std::string euid;
		switch (head.cmd & ~uint16_t(0x8000)){
			case 0:		// heartbeat
				return;
			case 1:		// lighting
				break;
			case 2:		// presure
				//ok = catch_call([&]{return process_hydrant(ip_addr_port, first, size, euid);});
				break;
			case 3:		// setting
				break;
			case 4:
				break;	// ask time
			case 5:
				//ok = catch_call([&]{return process_well_lid(ip_addr_port, first, size, euid);});
				break;	// ask time
			case 6:
				ok = catch_call([&]{return process_environment(ip_addr_port, first, size, euid, sender);});
				break;	// ask time
			case 7:
				//ok = catch_call([&]{return process_firedoor(ip_addr_port, first, size, euid);});
				break;
			case 8:
				//ok = catch_call([&]{return process_firebox(ip_addr_port, first, size, euid);});
				break;
			case 9:
				//ok = catch_call([&]{return process_sprinkler(ip_addr_port, first, size, euid);});
				break;
			case 10:
				ok = catch_call([&]{return process_voice(ip_addr_port, first, size, euid, sender);});
				break;
			case 11:
				//ok = catch_call([&]{return process_fire_pump(ip_addr_port, first, size, euid);});
				break;
			case 12:
				ok = catch_call([&]{return process_water_quality(ip_addr_port, first, size, euid, sender);});
				break;
			case 13:
				ok = catch_call([&]{return process_electric(ip_addr_port, first, size, euid, sender);});
				break;
			case 14:
				//ok = catch_call([&]{return process_device_init_response(ip_addr_port, first, size, euid);});
				break;
			case 15:
				//ok = catch_call([&]{return process_threshold_response(ip_addr_port, first, size, euid);});
				break;
			case 16:
				ok = catch_call([&]{return process_date_update(ip_addr_port, first, size, euid, sender);});
				break;
			case 17:
				ok = catch_call([&]{return process_xhs_electric(ip_addr_port, first, size, euid, sender);});
				break;
			case 19:
			case 20:
				//ok = catch_call([&]{return process_fire_pump__and_fan_info(ip_addr_port, first, size, euid, head.cmd);});
				break;
			case 21:
				//ok = catch_call([&]{return process_waterater_data(ip_addr_port, first, size, euid);});
				break;
			case 22:
				ok = catch_call([&]{return process_sponge_upconfig(ip_addr_port, first, size, euid, sender);});
				break;
			case 23:
				ok = catch_call([&]{return process_sponge_data(ip_addr_port, first, size, euid, sender);});
				break;
			case 0x1f:		// 31: soak
				//ok = catch_call([&]{return process_soak(ip_addr_port, first, size, euid);});
				break;
			default:
				ok = true;
				BOOST_LOG_TRIVIAL(debug) <<"Unknown command : 0x" << std::hex << head.cmd;
				break;
		}
		if (!ok)
			BOOST_LOG_TRIVIAL(debug) <<"process failed, command: 0x"<< std::hex << head.cmd;

// disable forward data via orgid and port
#if 0
		if (!euid.empty()){
			auto ptr = first - sizeof(HeadProtocol);
			std::vector<char> data(ptr, ptr + size + sizeof(HeadProtocol));
			auto self = shared_from_this();
			this->context.database->forwardData(euid, [self, data, head](const std::string& ip, const std::string& port){
					self->context.ios_aserver.post([self, data, ip, port, head]{
					  udp::resolver resolver(self->context.ios_aserver);
					  udp::resolver::query query(udp::v4(), ip, port);
					  boost::system::error_code ec;
					  auto itr = resolver.resolve(query, ec);
					  if (!ec) {
						  self->socket_forward.send_to(boost::asio::buffer(data, data.size()), *itr);
						  BOOST_LOG_TRIVIAL(debug) <<"Forward data to " << ip << ":" << port << " type:" << std::hex << uint16_t(head.cmd);
					  }
					  });
					});
		}
#endif

		if (!this->context.config.default_forward_host.empty()){
			std::vector<char> package(first - sizeof(HeadProtocol), first + size);
			auto self = shared_from_this();
			this->context.ios_aserver.post([self, package]{
					udp::resolver resolver(self->context.ios_aserver);
					udp::resolver::query query(udp::v4(), self->context.config.default_forward_host, self->context.config.default_forward_port);
					boost::system::error_code ec;
					auto itr = resolver.resolve(query, ec);
					if (!ec) {
					self->socket_forward.send_to(boost::asio::buffer(package, package.size()), *itr);
					BOOST_LOG_TRIVIAL(debug) <<"Forward data to " << self->context.config.default_forward_host;
					}
					});

		}
	}
	HeadType NodeProtocol::checkHead(const HeadData& h){
		if (h.head.sig == K_signature &&  h.head.version == 1){
			return HeadType::nbiot;
		}
		else if (h.head_fire.sig == K_sig_fire && h.head_fire.major_version == 1){
			return HeadType::fire;
		}

		return HeadType::invalid;
	}
	template<typename SizeType, typename Ar>
	std::string load_string(Ar& ar){
		size_t n = load<SizeType>(ar);
		std::vector<char> temp(n);
		block_read(ar.get(), to_range(temp));
		return std::string(temp.begin(), temp.end());
	}
	template<typename Ar>
	std::string load_small_string(Ar& ar){
		return load_string<uint8_t>(ar);
	}

	template<typename SizeType, typename Ar>
	Ar& save_string(Ar& ar, const std::string& s){
		ar & SizeType(s.size());
		if (!s.empty()){
			auto ptr = &*s.begin();
			block_write(ar.get(), make_range(ptr, ptr + s.size()));
		}
		return ar;
	}
	bool NodeProtocol::process_hydrant(const std::string& ip_addr_port, const char* buf, size_t size, std::string& euid){
		BOOST_LOG_TRIVIAL(debug) <<"process hydrant, size:"  << size ;
		buffer_in rd(range<const char*>(buf, buf + size));
		auto ar = s11n::as_source(rd);

		std::string sim_id;
		int32_t pressure = 0;
		int32_t trend = 0;
		uint8_t signal = 0;
		int32_t voltage = 0;

		while(rd.readable()){
			auto prop_id = load<uint16_t>(ar);
			switch(prop_id){
				case 1:		//euid
					euid = load_small_string(ar);
					break;
				case 20:	// pressure
					ar & pressure;
					break;
				case 28:	//
					ar & trend;
					break;
				case 29:	// signal
					ar & signal;
					break;
				case 30:	// sim_id
					sim_id = load_small_string(ar);
					break;
				case 23:	//voltage
					ar & voltage;
					break;
				default:
					return false;
			}
		}

		if (!euid.empty())
			context.ios_aserver.post([=]{context.database->saverHhydrant(ip_addr_port, euid, pressure, trend, uint16_t(signal), sim_id, voltage);});

		return true;
	}
	bool NodeProtocol::process_well_lid(const std::string& ip_addr_port, const char* buf, size_t size, std::string& euid){
		BOOST_LOG_TRIVIAL(debug) <<"process well lid, size:"  << size ;
		buffer_in rd(range<const char*>(buf, buf + size));
		auto ar = s11n::as_source(rd);

		std::string sim_id;
		uint8_t cover_status = 0;
		uint8_t submerged = 0xff;	// no data
		uint8_t signal = 0;

		while(rd.readable()){
			auto prop_id = load<uint16_t>(ar);
			switch(prop_id){
				case 1:		//euid
					euid = load_small_string(ar);
					break;
				case 29:	// signal
					ar & signal;
					break;
				case 30:	// sim_id
					sim_id = load_small_string(ar);
					break;
				case 33:	// cover status
					ar & cover_status;
					break;
				case 34:	// submerged
					ar & submerged;
					break;
				default:
					return false;
			}
		}

		if (!euid.empty())
			context.ios_aserver.post([=]{context.database->saveWellLip(ip_addr_port, euid, uint16_t(signal), sim_id, uint16_t(cover_status), uint16_t(submerged));});

		return true;
	}
	bool NodeProtocol::process_firedoor(const std::string& ip_addr_port, const char* buf, size_t size, std::string& euid){
		BOOST_LOG_TRIVIAL(debug) <<"process fire door, size:"  << size ;
		buffer_in rd(range<const char*>(buf, buf + size));
		auto ar = s11n::as_source(rd);

		std::string sim_id;
		uint8_t reed_status = 0;
		uint8_t submerged = 0xff;	// no data
		uint8_t signal = 0;
		int32_t voltage = 0;

		while(rd.readable()){
			auto prop_id = load<uint16_t>(ar);
			switch(prop_id){
				case 1:		//euid
					euid = load_small_string(ar);
					break;
				case 29:	// signal
					ar & signal;
					break;
				case 30:	// sim_id
					sim_id = load_small_string(ar);
					break;
				case 36:	// reed status
					ar & reed_status;
					break;
				case 34:	// submerged
					ar & submerged;
					break;
				case 23:	//voltage
					ar & voltage;
					break;
				default:
					return false;
			}
		}

		if (!euid.empty())
			context.ios_aserver.post([=]{context.database->saveFiredoor(ip_addr_port, euid, uint16_t(signal), sim_id, voltage, uint16_t(reed_status), uint16_t(submerged));});

		return true;
	}
	bool NodeProtocol::process_sprinkler(const std::string& ip_addr_port, const char* buf, size_t size, std::string& euid){
		BOOST_LOG_TRIVIAL(debug) <<"process sprinkler, size:"  << size ;
		buffer_in rd(range<const char*>(buf, buf + size));
		auto ar = s11n::as_source(rd);

		std::string sim_id;
		int32_t pressure = 0;
		uint8_t signal = 0;
		int32_t voltage = 0;

		while(rd.readable()){
			auto prop_id = load<uint16_t>(ar);
			switch(prop_id){
				case 1:		//euid
					euid = load_small_string(ar);
					break;
				case 20:	// pressure
					ar & pressure;
					break;
				case 29:	// signal
					ar & signal;
					break;
				case 30:	// sim_id
					sim_id = load_small_string(ar);
					break;
				case 23:	//voltage
					ar & voltage;
					break;
				default:
					return false;
			}
		}

		if (!euid.empty())
			context.ios_aserver.post([=]{context.database->saverSprinkler(ip_addr_port, euid, pressure, voltage, uint16_t(signal), sim_id);});

		return true;
	}
	bool NodeProtocol::process_firebox(const std::string& ip_addr_port, const char* buf, size_t size, std::string& euid){
		BOOST_LOG_TRIVIAL(debug) <<"process fire box, size:"  << size ;
		buffer_in rd(range<const char*>(buf, buf + size));
		auto ar = s11n::as_source(rd);

		std::string sim_id;
		int32_t water_deepth = 0;
		uint8_t signal = 0;
		int32_t voltage = 0;

		while(rd.readable()){
			auto prop_id = load<uint16_t>(ar);
			switch(prop_id){
				case 1:		//euid
					euid = load_small_string(ar);
					break;
				case 37:	// water deepth
					ar & water_deepth;
					break;
				case 29:	// signal
					ar & signal;
					break;
				case 30:	// sim_id
					sim_id = load_small_string(ar);
					break;
				case 23:	//voltage
					ar & voltage;
					break;
				default:
					return false;
			}
		}

		if (!euid.empty())
			context.ios_aserver.post([=]{context.database->saverFirebox(ip_addr_port, euid, water_deepth, voltage, uint16_t(signal), sim_id);});

		return true;
	}
	bool NodeProtocol::process_firepool(const std::string& ip_addr_port, const char* buf, size_t size, std::string& euid){
		xirang2::unused(ip_addr_port, buf, size);
		return true;
	}
	bool NodeProtocol::process_environment(const std::string& ip_addr_port, const char* buf, size_t size, std::string& euid, ResponseSenderType sender){
		BOOST_LOG_TRIVIAL(debug) <<"process environment, size:"  << size ;
		buffer_in rd(range<const char*>(buf, buf + size));
		auto ar = s11n::as_source(rd);

		std::string sim_id;
		uint8_t signal = 0;
		std::vector<Environment> env_data;
		uint16_t interval = 30;

		while(rd.readable()){
			auto prop_id = load<uint16_t>(ar);
			switch(prop_id){
				case 1:		euid = load_small_string(ar); break;
				case 29:	ar & signal; break;
				case 30:	sim_id = load_small_string(ar); break;
				case 32:	ar & interval; break;
				case 35:		// environment id
				{
					for(auto size = load<uint8_t>(ar); size > 0; --size){
						Environment env;
						ar & env.temperature & env.humidity & env.pm25 & env.pm10 & env.pm1 & env.wind_speed & env.wind_direction & env.noise; 
						env_data.push_back(env);
					}
					break;
				}
				default:
				return false;
			}
		}

		if (interval < 1) interval = 30;
		auto send_interval = [sender](uint16_t new_interval){
			std::vector<char> buf;
			buffer_out wr(buf);
			auto ar = s11n::as_sink(wr);
			ar & K_signature & K_version & K_reserved 
				& uint16_t(0x8006)
				& uint16_t(4)	// length 4
				& uint16_t(32)	// interval ID
				& new_interval;	// 
			sender(buf);
		};
		if (!euid.empty())
			context.ios_aserver.post([=]{context.database->saveEnvironment(ip_addr_port, euid, uint16_t(signal), sim_id, interval, env_data, send_interval);});

		return true;
	}
	bool NodeProtocol::process_voice(const std::string& ip_addr_port, const char* buf, size_t size, std::string& euid, ResponseSenderType sender){
		BOOST_LOG_TRIVIAL(debug) <<"process voice, size:"  << size ;
		buffer_in rd(range<const char*>(buf, buf + size));
		auto ar = s11n::as_source(rd);

		std::string sim_id;
		uint8_t signal = 0;
		uint16_t voice = 0;
		while(rd.readable()){
			auto prop_id = load<uint16_t>(ar);
			switch(prop_id){
				case 1:		//euid
					euid = load_small_string(ar);
					break;
				case 29:	// signal
					ar & signal;
					break;
				case 30:	// sim_id
					sim_id = load_small_string(ar);
					break;
				case 38:	// voice 
					voice = load<uint8_t>(ar);
					break;
				default:
					BOOST_LOG_TRIVIAL(debug) <<"Unknown property id "  << prop_id ;
					return false;
			}
		}
		auto send_voice = [sender](uint8_t new_voice){
			std::vector<char> buf;
			buffer_out wr(buf);
			auto ar = s11n::as_sink(wr);
			ar & K_signature & K_version & K_reserved 
				& uint16_t(0x800a)
				& uint16_t(4)	// length 4
				& uint16_t(38)	// voice
				& new_voice;	// 
			sender(buf);
		};
		if (!euid.empty())
			context.ios_aserver.post([=]{context.database->voiceHeartbeat(ip_addr_port, euid, uint16_t(signal), sim_id, voice, send_voice);});
		return true;
	}
	bool NodeProtocol::process_fire_pump(const std::string& ip_addr_port, const char* buf, size_t size, std::string& euid){
		BOOST_LOG_TRIVIAL(debug) <<"process fire pump, size:"  << size ;
		buffer_in rd(range<const char*>(buf, buf + size));
		auto ar = s11n::as_source(rd);

		std::string sim_id;
		uint8_t signal = 0;
		int32_t pressure = 0;
		int32_t voltage = 0;
		while(rd.readable()){
			auto prop_id = load<uint16_t>(ar);
			switch(prop_id){
				case 1:		//euid
					euid = load_small_string(ar);
					break;
				case 29:	// signal
					ar & signal;
					break;
				case 30:	// sim_id
					sim_id = load_small_string(ar);
					break;
				case 38:	// voice 
					pressure = load<int32_t>(ar);
					break;
				case 20:	// pressure
					ar & pressure;
					break;
				case 23:	//voltage
					ar & voltage;
					break;
				default:
					BOOST_LOG_TRIVIAL(debug) <<"Unknown property id "  << prop_id ;
					return false;
			}
		}
		if (!euid.empty())
			context.ios_aserver.post([=]{context.database->saverFirePump(ip_addr_port, euid, pressure, voltage, uint16_t(signal), sim_id);});
		return true;
	}
	bool NodeProtocol::process_water_quality(const std::string& ip_addr_port, const char* buf, size_t size, std::string& euid, ResponseSenderType sender){
		BOOST_LOG_TRIVIAL(debug) <<"process water quality, size:"  << size ;
		buffer_in rd(range<const char*>(buf, buf + size));
		auto ar = s11n::as_source(rd);

		std::string sim_id;
		uint8_t signal(0), move_value(0);
		uint16_t temperature = 0;
		uint16_t ntu(0), do_value(0), uvt(0);
		float ph(0), orp(0), ec_value(0), nh4(0), lat(0), longitude(0);
		
		while(rd.readable()){
			auto prop_id = load<uint16_t>(ar);
			switch(prop_id){
				case 1:		//euid
					euid = load_small_string(ar);
					break;
				case 29:	// signal
					ar & signal;
					break;
				case 30:	// sim_id
					sim_id = load_small_string(ar);
					break;
				case 26:	// temperature
					ar & temperature;
					break;
				case 40:	// ntu_value
					ar & ntu;
					break;
				case 41:	// ph value
					ar & ph;
					break;
				case 42:	// orp value
					ar & orp;
					break;
				case 43:	// ec value
					ar & ec_value;
					break;
				case 44:	// NH4 value
					ar & nh4;
					break;
				case 45:	// DO value
					ar & do_value;
					break;
				case 46:	// Mover value
					ar & move_value;
					break;
				case 47:	// Latitude
					ar & lat;
					break;
				case 48:	// Longitude
					ar & longitude;
					break;
				case 49:	// UVT
					ar & uvt;
					break;
				default:
					BOOST_LOG_TRIVIAL(debug) <<"Unknown property id "  << prop_id ;
					return false;
			}
		}
		if (!euid.empty())
			context.ios_aserver.post([=]{context.database->saverWaterQuality(ip_addr_port, euid, uint16_t(signal), sim_id, temperature, ntu, ph, orp, ec_value, nh4, do_value, move_value, lat, longitude, uvt);});
		return true;
	}
	bool NodeProtocol::process_electric(const std::string& ip_addr_port, const char* buf, size_t size, std::string& euid, ResponseSenderType sender){
		BOOST_LOG_TRIVIAL(debug) <<"process electric , size:"  << size ;
		buffer_in rd(range<const char*>(buf, buf + size));
		auto ar = s11n::as_source(rd);

		pcm_350a_data d{};

		while(rd.readable()){
			auto prop_id = load<uint16_t>(ar);
			switch(prop_id){
				case 1:		//euid
					d.euid = load_small_string(ar);
					euid = d.euid;
					break;
				case 29:	ar & d.signal; break;
				case 30:	d.simid = load_small_string(ar); break;
				case 0x101: ar & d.meter_voltage; break;
				case 0x102: ar & d.meter_current; break;
				case 0x103: ar & d.meter_current_phase; break;
				case 0x104: ar & d.frequency; break;
				case 0x105: ar & d.power_factor; break;
				case 0x106: ar & d.meter_3phase_power; break;
				case 0x107: ar & d.base_residual_current; break;
				case 0x108: ar & d.fullwave_residual_current; break;
				case 0x109: ar & d.di_value; break;
				case 0x10a: ar & d.do_value; break;
				case 0x10b: ar & d.meter_temperature; break;
				case 0x10c: ar & d.meter_self_test; break;
				case 0x10d: ar & d.total_harmonic_active_power; break;
				case 0x10e: ar & d.meter_energy; break;
				case 0x10f: ar & d.meter_current_thd; break;
				case 0x110: ar & d.meter_voltage_thd; break;
				case 0x111: ar & d.unbalance_factor; break;
				case 0x112: ar & d.date; break;
				case 0x113: ar & d.warning_code; break;
				default:
					BOOST_LOG_TRIVIAL(debug) <<"Unknown property id "  << prop_id ;
					return false;
			}
		}

		if (!euid.empty()){
			context.database->save_electric(ip_addr_port, d, sender);
		}
		return true;
	}
	bool NodeProtocol::process_device_init_response(const std::string& ip_addr_port, const char* buf, size_t size, std::string& euid){
		BOOST_LOG_TRIVIAL(debug) <<"process device init response, size:"  << size ;
		buffer_in rd(range<const char*>(buf, buf + size));
		auto ar = s11n::as_source(rd);

		int8_t signal = 0;
		std::string simid;
		int8_t result = 1;
		while(rd.readable()){
			auto prop_id = load<uint16_t>(ar);
			switch(prop_id){
				case 1:		euid = load_small_string(ar); break;
				case 29:	ar & signal; break;
				case 30:	simid = load_small_string(ar); break;
				case 0x118: ar & result;	break;
				default:
					BOOST_LOG_TRIVIAL(debug) <<"Unknown property id "  << prop_id ;
					return false;
			}
		}
		if (!euid.empty() && result == 0){
			context.database->clear_device_init_flag(euid);
		}

		return true;
	}
	bool NodeProtocol::process_threshold_response(const std::string& ip_addr_port, const char* buf, size_t size, std::string& euid){
		BOOST_LOG_TRIVIAL(debug) <<"process threshold response, size:"  << size ;
		buffer_in rd(range<const char*>(buf, buf + size));
		auto ar = s11n::as_source(rd);

		int8_t signal = 0;
		std::string simid;
		int8_t result = 1;
		while(rd.readable()){
			auto prop_id = load<uint16_t>(ar);
			switch(prop_id){
				case 1:		euid = load_small_string(ar); break;
				case 29:	ar & signal; break;
				case 30:	simid = load_small_string(ar); break;
				case 0x11a: ar & result;	break;
				default:
					BOOST_LOG_TRIVIAL(debug) <<"Unknown property id "  << prop_id ;
					return false;
			}
		}

		if (!euid.empty() && result == 0){
			context.database->clear_device_param_flag(euid);
		}
		return true;
	}
	bool NodeProtocol::process_date_update(const std::string& ip_addr_port, const char* buf, size_t size, std::string& euid, ResponseSenderType sender){
		BOOST_LOG_TRIVIAL(debug) <<"process data update, size:"  << size ;

		date_info date{};
		auto t = std::time(nullptr);
		std::tm tm{};
		localtime_r(&t, &tm);
		date.year = tm.tm_year - 100;	// tm year start from 1900, pcm date start from 2000
		date.month = tm.tm_mon + 1; // tm month start from 0, requires from 1
		date.day = tm.tm_mday;
		date.hour = tm.tm_hour;
		date.minute = tm.tm_min;
		date.second = tm.tm_sec;
		auto now = std::chrono::system_clock::now() - std::chrono::system_clock::time_point();
		date.millisecond = std::chrono::duration_cast<std::chrono::milliseconds>(now).count() %1000;

		{
			std::vector<char> response;
			buffer_out wr(response);
			auto ar = s11n::as_sink(wr);
			ar & K_signature & K_version & K_reserved 
				& uint16_t(0x10)
				& uint16_t(sizeof(date) + 2)
				& uint16_t(0x112)
				& date;

			sender(response);
		}

		return true;
	}

	template<typename T> static inline T get_value(std::istream& in){
		std::string field;
		std::getline(in, field, '|');
		return field.empty() ? 0 : boost::lexical_cast<T>(field);
	}

	template<typename T>
	std::vector<T> get_number_list(std::istream& in){
		std::string field;
		std::getline(in, field, '|');

		std::vector<T> ret;
		auto prev = field.begin();
		while(prev != field.end()){
			auto last = std::find(prev, field.end(), ',');
			std::string svalue(prev, last);
			ret.push_back(svalue.empty()? 0 : boost::lexical_cast<T>(svalue));
			prev = last;
			if (prev != field.end()) {
				++prev; //skip ','
			}
		}
		return ret;
	}

	std::vector<std::string> get_str_list(std::istream& in){
		std::string field;
		std::getline(in, field, '|');

		std::vector<std::string> ret;
		auto prev = field.begin();
		while(prev != field.end()){
			auto last = std::find(prev, field.end(), ',');
			ret.push_back(std::string(prev, last));
			prev = last;
			if (prev != field.end()) {
				++prev; //skip ','
			}
		}
		return ret;
	}

	static std::string g_ks = "xhscloud20180705";
	static CryptoPP::SecByteBlock g_key(reinterpret_cast<const unsigned char*>(g_ks.c_str()), g_ks.size());

	static std::string g_msg_start= "@@";
	static std::string g_msg_end = "##";

	std::string decrypt_msg(const std::string& sec_txt){
		std::string plain;
		CryptoPP::ECB_Mode< CryptoPP::AES >::Decryption d;
		d.SetKey( g_key, g_key.size() );
		CryptoPP::StringSource( sec_txt, true, 
				new CryptoPP::StreamTransformationFilter( d,
					new CryptoPP::StringSink( plain )
					)
				);
		return plain;
	}
	std::string encrypt_msg(const std::string& plain_txt){
		std::string sec_text;
		CryptoPP::ECB_Mode< CryptoPP::AES >::Encryption e;
		e.SetKey(g_key, g_key.size() );
		CryptoPP::StringSource( plain_txt, true, 
				new CryptoPP::StreamTransformationFilter( e,
					new CryptoPP::StringSink( sec_text ),
					CryptoPP::StreamTransformationFilter::PKCS_PADDING
					)
				);
		return sec_text;
	}
	std::string extract_xhs_euid(const std::string& s){
		auto pos = std::find(s.begin(), s.end(), '_');
		if (pos != s.end()) ++pos;
		pos = std::find(pos, s.end(), '_');
		if (pos != s.end()) ++pos;
		return std::string(pos, s.end());
	}

	std::function<void(const std::string&)> make_enc_sender(const std::string& euid, ResponseSenderType sender, bool faked){
		return [sender, euid, faked](const std::string& record){
			BOOST_LOG_TRIVIAL(debug) <<"send xhs record down:"  << record ;

			std::string sec_text = encrypt_msg(record);

			if (faked){
				sender(std::vector<char>(sec_text.begin(), sec_text.end()));
			}
			else 
			{
				uint16_t payload = uint16_t(
						2 +	// euid id
						1 + // euid size
						euid.size() + 
						2 + 	// message id
						2 +		// messahe size
						sec_text.size() // message size
						);

				std::vector<char> res2;
				buffer_out wr2(res2);
				auto ar2 = s11n::as_sink(wr2);
				ar2 & K_signature & K_version & K_reserved 
					& uint16_t(17)	// xhs command
					& payload	// total length
					& uint16_t(1);			// euid
				save_string<uint8_t>(ar2, euid);
				ar2 & uint16_t(0x32);		// message
				save_string<uint16_t>(ar2, sec_text);

				sender(res2);
			}
		};

	}

	bool NodeProtocol::process_xhs_electric(const std::string& ip_addr_port, const char* buf, size_t size, std::string& euid, ResponseSenderType sender){
		BOOST_LOG_TRIVIAL(debug) <<"process xhs electric, size:"  << size ;
		buffer_in rd(range<const char*>(buf, buf + size));
		auto ar = s11n::as_source(rd);

		int8_t signal_ext = 0;
		std::string simid_ext, message;
		while(rd.readable()){
			auto prop_id = load<uint16_t>(ar);
			switch(prop_id){
				case 1:		euid = load_small_string(ar); break;
				case 29:	ar & signal_ext; break;
				case 30:	simid_ext = load_small_string(ar); break;
				case 0x32:  message = load_string<uint16_t>(ar);	break;
				default:
					BOOST_LOG_TRIVIAL(debug) <<"Unknown property id "  << prop_id ;
					return false;
			}
		}

		if (message.empty())
			return false;

		bool is_faked_package = euid.empty();

		auto plain = decrypt_msg(message);
		BOOST_LOG_TRIVIAL(debug) <<"Received xhs record:"  << plain ;

		auto start = std::search(plain.begin(), plain.end(), g_msg_start.begin(), g_msg_start.end()); 
		if (start == plain.end()) return false;
	
		start += g_msg_start.size();
		auto end = std::search(start, plain.end(), g_msg_end.begin(), g_msg_end.end()); 

		std::string field;
		std::stringstream sstr(std::string(start, end));
		std::getline(sstr, field, '|');
		int msg_id = boost::lexical_cast<int>(field);

		std::getline(sstr, field, '|');
		int msg_serial = boost::lexical_cast<int>(field);

		switch (msg_id){
			case 0:		//heard beat
				break;	// do nothing
			case 1:		// message acknowledge
				break;	// do nothing
			case 2:		// message deny
				break;	// do nothing
			case 3:		// bind module info
				{
					std::string simid, skip, euid_raw;
					for(int i =0; i < 3;++i)
						std::getline(sstr, skip, '|');
					std::getline(sstr, simid, '|');
					std::getline(sstr, euid_raw, '|');
					euid = extract_xhs_euid(euid_raw);
					context.database->update_sim_id(euid, simid, ip_addr_port);
				}
				break;	// 
			case 4:		// send operation
				break;
			case 5:		// send device status
				break;
			case 6:		// send analog info
				{
					std::string euid_raw;
					std::getline(sstr, euid_raw, '|');
					xhs_send_anolog record;
					euid = extract_xhs_euid(euid_raw);
					record.module = euid;
					record.circuit = get_value<uint32_t>(sstr);
					record.device_addr = get_value<uint32_t>(sstr);
					record.channel = get_number_list<uint32_t>(sstr);
					record.device_code = get_value<uint32_t>(sstr);
					record.flag = get_value<uint32_t>(sstr);
					record.vender = get_value<uint32_t>(sstr);
					record.device_type = get_number_list<uint32_t>(sstr);
					record.anolog_value = get_number_list<double>(sstr);
					record.device_status = get_number_list<uint32_t>(sstr);
					record.signal = get_value<uint32_t>(sstr);
					context.database->save_anolog(ip_addr_port, record, make_enc_sender(euid, sender, is_faked_package));
				}
				break;
			case 7:		// send analog configuration
				{
					std::string euid_raw;
					std::getline(sstr, euid_raw, '|');

					xhs_send_anolog_config record;
					euid = extract_xhs_euid(euid_raw);
					record.module = euid;
					record.circuit = get_value<uint32_t>(sstr);
					record.device_addr = get_value<uint32_t>(sstr);
					record.channel = get_number_list<uint32_t>(sstr);
					record.device_code = get_value<uint32_t>(sstr);
					record.flag = get_value<uint32_t>(sstr);
					record.vender = get_value<uint32_t>(sstr);
					record.device_type = get_number_list<uint32_t>(sstr);
					record.signal_type = get_number_list<uint32_t>(sstr);
					record.low_value = get_number_list<uint32_t>(sstr);
					record.high_value = get_number_list<uint32_t>(sstr);
					record.unit = get_str_list(sstr);

					context.database->save_anolog_config(ip_addr_port, record, make_enc_sender(euid, sender, is_faked_package));
				}
				break;
			case 9:		// confirm DTU config info	
				break;	// do nothing
			default:
				break;
		}

		BOOST_LOG_TRIVIAL(debug) <<" xhs record end " ;

		std::stringstream sout;
		sout << g_msg_start << 1 << '|' << msg_serial << g_msg_end; // 1: confirm message
		make_enc_sender(euid, sender, is_faked_package)(sout.str());

		
		return true;
	}
	bool NodeProtocol::process_fire_pump__and_fan_info(const std::string& ip_addr_port, const char* buf, size_t size, std::string& euid, uint16_t cmd){
		BOOST_LOG_TRIVIAL(debug) <<"process_fire_pump__and_fan_info, size:"  << size ;
		buffer_in rd(range<const char*>(buf, buf + size));
		auto ar = s11n::as_source(rd);
		info_fire_pump_information data;
		data.cmd_id = cmd;
		data.ip_addr_port = ip_addr_port;

		while(rd.readable()){
			auto prop_id = load<uint16_t>(ar);
			switch(prop_id){
				case 1:		//euid
					data.euid = load_small_string(ar);
					break;
				case 29:	// signal
					data.signal = load<uint8_t>(ar);
					break;
				case 30:	// sim_id
					data.sim_id = load_small_string(ar);
					break;
				case 23:	//voltage
					ar & data.voltage;
					break;
				case 0x34:	//AC_Det
					{
						auto arr = load_small_string(ar);
						for(auto c : arr)
							data.ac_det.push_back(c== 0? '0' : (c == 1 ? '1' : '2'));
						std::reverse(arr.begin(), arr.end());
					}
					break;
				default:
					BOOST_LOG_TRIVIAL(debug) <<"process_fire_pump__and_fan_info, error prop_id:"  << size ;
					return false;
			}
		}
		euid = data.euid;

		context.database->save_fire_pump_and_fan_info(data);

		return true;
	}
	bool NodeProtocol::process_waterater_data(const std::string& ip_addr_port, const char* buf, size_t size, std::string& euid){
		BOOST_LOG_TRIVIAL(debug) <<"process_waterater_data, size:"  << size ;
		buffer_in rd(range<const char*>(buf, buf + size));
		auto ar = s11n::as_source(rd);
		info_waterater_data data;
		data.ip_addr_port = ip_addr_port;

		while(rd.readable()){
			auto prop_id = load<uint16_t>(ar);
			switch(prop_id){
				case 1:		//euid
					data.euid = load_small_string(ar);
					break;
				case 29:	// signal
					data.signal = load<uint8_t>(ar);
					break;
				case 30:	// sim_id
					data.sim_id = load_small_string(ar);
					break;
				case 23:	//voltage
					ar & data.voltage;
					break;
				case 0x35:	//AC_Det
					{
						ar & data.velocity & data.positive_cumulative_flow & data.negative_cumulative_flow
							& data.net_cumulative_flow & data.error_code;
						data.work_steps = load<uint8_t>(ar);
						data.upstream_signal = load<uint8_t>(ar);
						data.downstream_signal = load<uint8_t>(ar);
					}
					break;
				default:
					BOOST_LOG_TRIVIAL(debug) <<"process_waterater_data, error prop_id:"  << size ;
					return false;
			}
		}
		euid = data.euid;

		context.database->save_waterater_data(data);

		return true;
	}

	bool NodeProtocol::process_sponge_data(const std::string& ip_addr_port, const char* buf, size_t size, std::string& euid, ResponseSenderType sender){
		BOOST_LOG_TRIVIAL(debug) <<"process_sponge_data, size:"  << size ;
		buffer_in rd(range<const char*>(buf, buf + size));
		auto ar = s11n::as_source(rd);
		sponge_data data;
		data.ip_addr_port = ip_addr_port;

		while(rd.readable()){
			auto prop_id = load<uint16_t>(ar);
			switch(prop_id){
				case 1:		//euid
					data.euid = load_small_string(ar);
					break;
				case 29:	// signal
					data.signal = load<uint8_t>(ar);
					break;
				case 30:	// sim_id
					data.sim_id = load_small_string(ar);
					break;
				case 23:	//voltage
					ar & data.voltage;
					break;
				case 0x36:	//sensor cond
					for(size_t i =0; i < 4; ++i)
						ar & data.sensor_cond[i];
					break;
				case 0x3f:  //flow meter
					ar & data.flow.flow & data.flow.flow_all & data.flow.temperature & data.flow.rate & data.flow.depth;
					break;
				case 0x37:  //ss meter
					ar & data.ss.ss & data.ss.factor & data.ss.humidity;
					break;
				case 0x38: //COD meter
					ar & data.cod.cod & data.cod.turbidity;
					break;
				case 0x39: //rain meter
					ar & data.rain.hyetal & data.rain.hyetal_all;
					break;
				default:
					BOOST_LOG_TRIVIAL(debug) <<"process_sponge_data, error prop_id:"  << prop_id ;
					return false;
			}
		}

		euid = data.euid;
		context.database->save_sponge_data(data, sender);
		return true;
	}
	bool NodeProtocol::process_sponge_upconfig(const std::string& ip_addr_port, const char* buf, size_t size, std::string& euid, ResponseSenderType sender){
		BOOST_LOG_TRIVIAL(debug) <<"process_sponge_upconfig, size:"  << size ;
		buffer_in rd(range<const char*>(buf, buf + size));
		auto ar = s11n::as_source(rd);
		sponge_upconfig data;
		data.ip_addr_port = ip_addr_port;

		while(rd.readable()){
			auto prop_id = load<uint16_t>(ar);
			switch(prop_id){
				case 1:		//euid
					data.euid = load_small_string(ar);
					break;
				case 29:	// signal
					data.signal = load<uint8_t>(ar);
					break;
				case 30:	// sim_id
					data.sim_id = load_small_string(ar);
					break;
				case 23:	//voltage
					ar & data.voltage;
					break;
				case 0x3a:	//Sys Time Condition
					for(size_t i =0; i < 4; ++i)
						ar & data.sys_time_cond[i];
					break;
				case 0x3b:
					ar & data.flow_cond[0] & data.flow_cond[1];
					break;
				case 0x3c:
					ar & data.cod_cond[0] & data.cod_cond[1];
					break;
				case 0x50:
					data.flow_cal_con = load<uint8_t>(ar);
					break;
				default:
					BOOST_LOG_TRIVIAL(debug) <<"process_sponge_upconfig, error prop_id:"  << prop_id ;
					return false;
			}
		}

		euid = data.euid;
		context.database->save_sponge_upconfig(data, sender);
		return true;
	}
	bool NodeProtocol::process_soak(const std::string& ip_addr_port, const char* buf, size_t size, std::string& euid){
		BOOST_LOG_TRIVIAL(debug) <<"process hydrant, size:"  << size ;
		buffer_in rd(range<const char*>(buf, buf + size));
		auto ar = s11n::as_source(rd);

		std::string sim_id;
		int32_t soak = 0;
		int32_t trend = 0;
		uint8_t signal = 0;
		int32_t voltage = 0;

		while(rd.readable()){
			auto prop_id = load<uint16_t>(ar);
			switch(prop_id){
				case 1:		//euid
					euid = load_small_string(ar);
					break;
				case 0x49:	// soak
					ar & soak;
					break;
				case 28:	//
					ar & trend;
					break;
				case 29:	// signal
					ar & signal;
					break;
				case 30:	// sim_id
					sim_id = load_small_string(ar);
					break;
				case 23:	//voltage
					ar & voltage;
					break;
				default:
					return false;
			}
		}

		if (!euid.empty())
			context.ios_aserver.post([=]{context.database->saveSoak(ip_addr_port, euid, soak, trend, uint16_t(signal), sim_id, voltage);});

		return true;
	}
}
